package com.bw.day0629;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Random;

/*
1.获取ods中日志数topic_log
2.处理过滤topic_log
3.新用户进行处理
4.把主流数据拆分成5个流分别存在5个主题里
    1.创建流式环境
    2.设置并行度
    3.运行jar向topic_log主题发送数据
    4.数据清洗，转换结构
    5.将脏数据写到kfk指定主题
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //todo 1 初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(4);
        //todo 2启动状态后端
        // 配置 Flink 状态后端（如使用 HashMapStateBackend）、Checkpoint（检查点，保障故障恢复） 重启策略（任务失败后重试逻辑 ）
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies
//                .failureRateRestart(10,
//                        Time.of(3L, TimeUnit.DAYS),
//                        Time.of(1L, TimeUnit.MINUTES)));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "bw");
        //todo 3 从kfk读取主流数据
        String topic = "topic_log";
        // 消费者组 ID
        //base_log_consumer 处理基础日志数据的消费者组
        String groupId = "base_log_consumer";
        // 将kfk接入为flink数据源 source 是原始日志字符串流
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
//        source.print();
        //todo 4 数据清洗，转换结构
         //todo 4.1 定义错误侧输出流
        OutputTag<String> dirtyStreamTag = new OutputTag<String>("dirtyStream") {
        };
        SingleOutputStreamOperator<String> cleanedStream = source.process(
                new ProcessFunction<String, String>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<String> out) throws Exception {
                        try {
                            //判断JSON
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            //发送主流
                            out.collect(jsonStr);
                        } catch (Exception e) {
                            //解析失败，发送到侧输出流（脏数据）
                            ctx.output(dirtyStreamTag, jsonStr.toString());
                            //dirtyStreamTag以他输出的就是脏数据
                        }
                    }
                }
        );
         //todo 4.2将脏数据写到kfk指定主题
        // 获取侧输出流（脏数据）
        DataStream<String> dirtyStream = cleanedStream.getSideOutput(dirtyStreamTag);
        //脏数据主题名
        String dirtyTopic = "dirty_data";
        //将脏数据写入到kfk
        dirtyStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(dirtyTopic));
//        dirtyStream.print();
        //todo 4.3 转换主流数据结构 jsonStr -> jsonObj
        SingleOutputStreamOperator<JSONObject> mappedStream = cleanedStream.map(JSONObject::parseObject);

    //todo 5 新老访客状态标记修复
        //todo 5.1按照mid对数据进行分组
        KeyedStream<JSONObject, String> keyedStream
                = mappedStream.keyBy(r -> r.getJSONObject("common").getString("mid"));
        //todo 5.2新访客状态标记修复
        // 状态
        SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            // 维护首次访问日期
            ValueState<String> firstViewDtState;
            // ValueState 用于存储单个值
            @Override
            //是初始化 ValueState
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                // getRuntimeContext()获取上下文
                // getState 通过状态描述从运行时上下文获取 ValueState 实例（firstViewDtState）
                firstViewDtState  = getRuntimeContext().getState(new ValueStateDescriptor<String>(
                //       状态名称         状态存储类型
                        "lastLoginDt", String.class
                ));
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                String isNew = jsonObject.getJSONObject("common").getString("is_new");
                String firstViewDt = firstViewDtState.value();
                Long ts = jsonObject.getLong("ts");
                String dt = DateFormatUtil.toDate(ts);
                // 如果等于1 并且状态为空更新dt首次访问时间
                if ("1".equals(isNew)) {
                    if (firstViewDt == null) {
                        firstViewDtState.update(dt);
                    } else {
                        // 反之 状态非空，日期不等于dt 是老访客
                        if (!firstViewDt.equals(dt)) {
                            isNew = "0";
                            // 修改
                            jsonObject.getJSONObject("common").put("is_new", isNew);
                        }
                    }
                } else {
                    // 如果状态为空
                    if (firstViewDt == null) {
                        //将首次访问日期设置为昨天
                        String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                        firstViewDtState.update(yesterday);
                    }
                }
                collector.collect(jsonObject);
            }

            @Override
            public void close() throws Exception {
                super.close();
            }
        });

    //todo 6 分流
        // todo 6.1 定义启动、曝光、动作、错误侧输出流
        OutputTag<String> startTag = new OutputTag<String>("startTag") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {
        };
        OutputTag<String> errorTag = new OutputTag<String>("errorTag") {
        };
        //todo 6.2分流
        SingleOutputStreamOperator<String> separatedStream = fixedStream.process(new ProcessFunction<JSONObject, String>() {

            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                //todo 6.2.1 收集错误数据
                JSONObject error = jsonObject.getJSONObject("err");
                if (error != null) {
                    context.output(errorTag, jsonObject.toJSONString());
                }
                //剔除 'err' 字段
                jsonObject.remove("err");
                //todo 6.2.2 收集启动数据
                JSONObject start = jsonObject.getJSONObject("start");
                if (start != null) {
                    context.output(startTag, jsonObject.toJSONString());
                } else {
                    //todo 6.2.2/1 获取 'page' 字段
                    JSONObject page = jsonObject.getJSONObject("page");
                    //todo 6.2.2/1 获取 'common' 字段
                    JSONObject common = jsonObject.getJSONObject("common");
                    //todo 6.2.2/1 获取 'ts' 字段
                    Long ts = jsonObject.getLong("ts");
                    // todo 6.2.3 收集曝光数据
                    JSONArray displays = jsonObject.getJSONArray("displays");
                    if (displays != null) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            JSONObject displayObj = new JSONObject();
                            displayObj.put("display", display);
                            displayObj.put("common", common);
                            displayObj.put("page", page);
                            displayObj.put("ts", ts);
                            context.output(displayTag, displayObj.toJSONString()
                            );
                        }
                    }
                    // todo 6.2.4 收集动作数据
                    JSONArray actions = jsonObject.getJSONArray("actions");
                    if (actions != null) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            JSONObject actionObj = new JSONObject();
                            actionObj.put("action", action);
                            actionObj.put("common", common);
                            actionObj.put("page", page);
                            context.output(actionTag, actionObj.toJSONString());
                        }
                    }
                    //todo 6.2.5 收集页面数据
                    jsonObject.remove("display");
                    jsonObject.remove("actions");
                    collector.collect(jsonObject.toJSONString());
                }
            }
        });
        // 打印主流和各侧输出流查看分流效果
        separatedStream.getSideOutput(startTag).print("start>>>>>>>>>>>>");
        separatedStream.getSideOutput(displayTag).print("displays>>>>>>>>>>>>");
        separatedStream.getSideOutput(actionTag).print("actions>>>>>>>>>>>>");
        separatedStream.getSideOutput(errorTag).print("error>>>>>>>>>>>>");
        separatedStream.print("page>>>>>>>>>>>>");

    //todo 7 将数据输出到kfk主题
        //todo 7.1提取各侧输出流
        DataStream<String> startDS = separatedStream.getSideOutput(startTag);
        DataStream<String> displayDS = separatedStream.getSideOutput(displayTag);
        DataStream<String> actionDS = separatedStream.getSideOutput(actionTag);
        DataStream<String> errorDS = separatedStream.getSideOutput(errorTag);
        //todo 7.2 定义不同日志输出到kfk的主题名称
        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";
        // page 反之
        separatedStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));

        env.execute();
    }
}